package com.chencong.transform;

import com.chencong.env.FlinkTableEnv;
import com.chencong.udf.MyFlatMapFunc;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.*;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;

/**
 * @Author chencong
 * @Description 批处理demo
 * @Date 8:25 下午 2021/8/4
 * @Param
 **/
public class Flink01BatchWordCount {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        ExecutionEnvironment batchTableEnvironment = FlinkTableEnv.getBatchTableEnvironment();


        // 2. 从文件读取数据  按行读取(存储的元素就是每行的文本)
        DataSource<String> lineData = batchTableEnvironment.readTextFile("/Users/chencong/IdeaProjects/BigData_Learning/flink/input");


        // 3. 转换数据格式
        FlatMapOperator<String, String> wordDS = lineData.flatMap(new MyFlatMapFunc());
        //转为元祖
        MapOperator<String, Tuple2<String, Integer>> wordToOne = wordDS.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
                return Tuple2.of(s, 1);
            }
        });


        // 4. 按照 word 进行分组 (批处理用group by)
        UnsortedGrouping<Tuple2<String, Integer>> tuple2UnsortedGrouping = wordToOne.groupBy(0);


        // 5. 分组内聚合统计
        AggregateOperator<Tuple2<String, Integer>> result = tuple2UnsortedGrouping.sum(1);

        // 6. 打印结果
        result.print();

        // 7. 批处理不用执行 流式任务才需要开启
        //batchTableEnvironment.execute();
    }
}
